package com.corn.flink.lesson6;

import cn.hutool.core.collection.CollUtil;
import com.corn.flink.lesson4.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/**
 * @author JimWu
 * @date 2023/3/24 10:17
 **/
public class WindowProcessFunction {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
        List<String> eventType = CollUtil.newArrayList("add", "delete", "update", "select");
        DataStreamSource<Event> ds = env.addSource(new RichSourceFunction<Event>() {
            private boolean flag = true;

            @Override
            public void run(SourceContext<Event> ctx) throws Exception {
                while (flag) {
                    ctx.collect(new Event(
                            eventType.get(ThreadLocalRandom.current().nextInt(eventType.size())),
                            System.currentTimeMillis()
                    ));
                    TimeUnit.SECONDS.sleep(1);
                }
            }

            @Override
            public void cancel() {
                this.flag = false;
            }
        });
        ds.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }))
                .keyBy(data -> data.type)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<Event, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<Event> elements, Collector<String> out) throws Exception {

                    }


                })
                .print();

        env.execute();
    }
}
